package com.haiyou.data.common.accessor.redis.redisson.mapping;

import java.io.Serializable;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Function;

import org.redisson.api.MapCacheOptions;
import org.redisson.api.MapOptions;
import org.redisson.api.RMap;
import org.redisson.api.RMapCache;

import com.google.common.collect.Iterables;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.haiyou.common.entity.Entity;
import com.haiyou.data.common.accessor.NotImplementedException;
import com.haiyou.data.common.cache.redis.RKeys;
import com.haiyou.data.common.cache.redis.redisson.RedissonAtomicExecutor;

/**
 * 依赖玩家id作为节点划分redisson映射
 * 
 * @author hong
 *
 * @param <K>
 * @param <V>
 */
public class KeySlotMappingAccessor<K extends Serializable & Comparable<K>, V extends Entity<K>>
		extends MappingAccessor<K, V> {

	private static final String SUFFIX = "MappingView_KeySlot";

	protected RMapCache<K, V> createMapCache() {
		MapCacheOptions<K, V> mapOptions = MapCacheOptions.defaults();
		// mapOptions.loader(mapLoader);
		mapOptions.writer(mapWriter);
		return new RedissonMapCacheWrapper<>(redissonClient, keyClass,
				RKeys.createRKey(RMapCache.class, keyPrefix, SUFFIX), mapOptions, nodeSize);
	}

	private RMap<K, V> getMapping(String playerId) {
		if (mapping instanceof RedissonMapCacheWrapper) {
			return ((RedissonMapCacheWrapper<K, V>) mapping).getMapping(playerId);
		}
		return mapping;
	}

	/**
	 * 注意: 此处不调用 synchronizer.flush(K,V)
	 */
	public Map<K, V> getAll(String playerId, Iterable<? extends K> keys) {
		// log.info("getAll keys:{}",keys);
		// redis中数据

		Set<K> keySet = Sets.newHashSet(keys);

		Map<K, V> maps = getMapping(playerId).getAll(Sets.newHashSet(keys));

		// 移除redis中 已有
		keySet.removeAll(maps.keySet());

		if (!keySet.isEmpty()) {
			// 查询数据库
			maps.putAll(queryAll(playerId, keySet));
		}

		return maps;
	}

	@SuppressWarnings("unchecked")
	public boolean containsKey(String playerId, Object key) {
		return getMapping(playerId).containsKey(key) || persister.contains((K) key);
	}

	@SuppressWarnings("unchecked")
	public boolean containsValue(String playerId, Object value) {
		return get(playerId, ((V) value).getId()).equals(value);
	}

	@SuppressWarnings("unchecked")
	public V get(String playerId, Object key) {
		V v = null;
		try {
			v = getMapping(playerId).get(key);

			if (v == null) {
				v = query(playerId, (K) key);
				return v;
			}
			return v;
		} finally {
//			if (v != null) {
//				synchronizer.flush((K) key, v);
//			}
		}
	}

	/**
	 * 通过数据库查询
	 * 
	 * @param key
	 * @return
	 */
	public V query(String playerId, K key) {

		V v = persister.query(key);

		if (v != null) {
			put(playerId, key, v);
		}
		return v;
	}

	/**
	 * 批量查询
	 * 
	 * @param keys
	 * @return
	 */
	public Map<K, V> queryAll(String playerId, Set<K> keys) {

		List<V> vs = persister.queryBatch(keys);

		Map<K, V> map = Maps.newHashMap();

		for (V v : vs) {
			Entity<K> entity = (Entity<K>) v;
			map.put(entity.getId(), v);
			put(playerId, entity.getId(), v);
		}

		return map;
	}

	public V put(String playerId, K key, V value) {
		RMap<K, V> map = getMapping(playerId);
		if (timeToLive > 0 && map instanceof RMapCache) {
			RMapCache<K, V> mapCache = (RMapCache<K, V>) map;
			return mapCache.put(key, value, timeToLive, timeToLiveUnit);
		} else {
			return map.put(key, value);
		}
	}

	public V remove(String playerId, Object key) {
		return getMapping(playerId).remove(key);
	}

	public void removeAll(String playerId, Iterable<? extends K> keys) {
		getMapping(playerId).fastRemove(Iterables.toArray(keys, keyClass));
		synchronizer.giveUp(keys);
		persister.delete(keys);
		if (keyRecycler != null)
			keyRecycler.retrieve(Iterables.toArray(keys, keyClass));
	}

	public void putAll(String playerId, Map<? extends K, ? extends V> m) {
		RMap<K, V> map = getMapping(playerId);
		if (timeToLive > 0 && map instanceof RMapCache) {
			RMapCache<K, V> mapCache = (RMapCache<K, V>) map;
			mapCache.putAll(m, timeToLive, timeToLiveUnit);
		} else {
			// RMapCache 中 putAll(m)方法为final不能重写
			// 修改成调用可以重写的putAll(m,size),重写后继续调用putAll(m)接口
			map.putAll(m, m.size());
		}
	}

	public V putIfAbsent(String playerId, K key, V value) {
		RMap<K, V> rmap = getMapping(playerId);
		return RedissonAtomicExecutor.call(rmap, key, lockWaitTime(), lockLeaseTime(), lockTimeUnit(),
				() -> rmap.putIfAbsent(key, value));
	}

	@SuppressWarnings("unchecked")
	public boolean remove(String playerId, Object key, Object value) {
		AtomicBoolean bool = new AtomicBoolean();
		RMap<K, V> rmap = getMapping(playerId);
		RedissonAtomicExecutor.run(rmap, (K) key, lockWaitTime(), lockLeaseTime(), lockTimeUnit(),
				() -> bool.compareAndSet(false, rmap.remove(key, value)));
		return bool.get();
	}

	public boolean replace(String playerId, K key, V oldValue, V newValue) {
		RMap<K, V> rmap = getMapping(playerId);
		AtomicBoolean bool = new AtomicBoolean();
		RedissonAtomicExecutor.run(getMapping(playerId), key, lockWaitTime(), lockLeaseTime(), lockTimeUnit(),
				() -> bool.compareAndSet(false, rmap.replace(key, oldValue, newValue)));
		return bool.get();
	}

	public V replace(String playerId, K key, V value) {
		RMap<K, V> rmap = getMapping(playerId);
		return RedissonAtomicExecutor.call(getMapping(playerId), key, lockWaitTime(), lockLeaseTime(), lockTimeUnit(),
				() -> rmap.replace(key, value));
	}

	public V computeIfAbsent(String playerId, K key, Function<? super K, ? extends V> mappingFunction) {
		RMap<K, V> rmap = getMapping(playerId);
		return RedissonAtomicExecutor.call(getMapping(playerId), key, lockWaitTime(), lockLeaseTime(), lockTimeUnit(),
				() -> rmap.computeIfAbsent(key, mappingFunction));
	}

	public V computeIfPresent(String playerId, K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction) {
		RMap<K, V> rmap = getMapping(playerId);
		return RedissonAtomicExecutor.call(getMapping(playerId), key, lockWaitTime(), lockLeaseTime(), lockTimeUnit(),
				() -> rmap.computeIfPresent(key, remappingFunction));
	}

	public V compute(String playerId, K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction) {
		RMap<K, V> rmap = getMapping(playerId);
		return RedissonAtomicExecutor.call(getMapping(playerId), key, lockWaitTime(), lockLeaseTime(), lockTimeUnit(),
				() -> rmap.compute(key, remappingFunction));
	}

	public V merge(String playerId, K key, V value, BiFunction<? super V, ? super V, ? extends V> remappingFunction) {
		RMap<K, V> rmap = getMapping(playerId);
		return RedissonAtomicExecutor.call(getMapping(playerId), key, lockWaitTime(), lockLeaseTime(), lockTimeUnit(),
				() -> rmap.merge(key, value, remappingFunction));
	}

	public V generate(String playerId, Consumer<V> consumer) {
		K k = null;
		V v = null;
		try {
			k = generateKey();
			v = valueClass.newInstance();
			v.setId(k);
			if (consumer != null)
				consumer.accept(v);
			put(playerId, k, v);
		} catch (Exception e) {
			throw new RuntimeException(e.getMessage(), e);
		}
		return v;
	}

	@Override
	@Deprecated
	public Map<K, V> getAll(Iterable<? extends K> keys) {
		throw new NotImplementedException();
	}

	@Override
	@Deprecated
	public boolean containsKey(Object key) {
		throw new NotImplementedException();
	}

	@Override
	@Deprecated
	public boolean containsValue(Object value) {
		throw new NotImplementedException();
	}

	@Override
	@Deprecated
	public V get(Object key) {
		throw new NotImplementedException();
	}

	/**
	 * 通过数据库查询
	 * 
	 * @param key
	 * @return
	 */
	@Deprecated
	public V query(K key) {
		throw new NotImplementedException();
	}

	/**
	 * 批量查询
	 * 
	 * @param keys
	 * @return
	 */
	@Deprecated
	public Map<K, V> queryAll(Set<K> keys) {
		throw new NotImplementedException();
	}

	@Override
	@Deprecated
	public V put(K key, V value) {
		throw new NotImplementedException();
	}

	@Override
	@Deprecated
	public V remove(Object key) {
		throw new NotImplementedException();
	}

	@Override
	@Deprecated
	public void removeAll(Iterable<? extends K> keys) {
		throw new NotImplementedException();
	}

	@Override
	@Deprecated
	public void putAll(Map<? extends K, ? extends V> m) {
		throw new NotImplementedException();
	}

	@Override
	@Deprecated
	public V putIfAbsent(K key, V value) {
		throw new NotImplementedException();
	}

	@Override
	@Deprecated
	public boolean remove(Object key, Object value) {
		throw new NotImplementedException();
	}

	@Override
	@Deprecated
	public boolean replace(K key, V oldValue, V newValue) {
		throw new NotImplementedException();
	}

	@Override
	@Deprecated
	public V replace(K key, V value) {
		throw new NotImplementedException();
	}

	@Override
	@Deprecated
	public V computeIfAbsent(K key, Function<? super K, ? extends V> mappingFunction) {
		throw new NotImplementedException();
	}

	@Override
	@Deprecated
	public V computeIfPresent(K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction) {
		throw new NotImplementedException();
	}

	@Override
	@Deprecated
	public V compute(K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction) {
		throw new NotImplementedException();
	}

	@Override
	@Deprecated
	public V merge(K key, V value, BiFunction<? super V, ? super V, ? extends V> remappingFunction) {
		throw new NotImplementedException();
	}

	@Override
	public Class<?> getInitClass() {
		return this.getClass();
	}
}
